package bigdata.hourglass.utils;

import bigdata.hourglass.common.domain.Bean;
import bigdata.hourglass.common.utils.ZkUtils;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import org.I0Itec.zkclient.ZkClient;
import scala.collection.Iterator;
import scala.collection.Seq;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;

/**
 * @author leo.jie (weixiao.me@aliyun.com)
 * @version 1.0
 * @organization CC
 * @website https://www.jlpyyf.com
 * @date 2019/12/8 7:23 下午
 * @since 1.0
 */
@Deprecated
public class AkkaUtils {
    public static Config getConfig(ZkClient zkClient) {
        String ip = "127.0.0.1";
        try {
            ip = InetAddress.getLocalHost().getHostAddress();
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
        Seq<Bean.HourglassEngine> allEngineInCluster = ZkUtils.getAllEngineInCluster(zkClient);
        Iterator<Bean.HourglassEngine> iterator = allEngineInCluster.iterator();
        List<Integer> ids = new ArrayList<>();
        List<Integer> ports = new ArrayList<>();
        while (iterator.hasNext()) {
            Bean.HourglassEngine engine = iterator.next();
            ids.add(engine.engineId());
            if (engine.engineInfo().contains(ip)) {
                ports.add(Integer.valueOf(engine.engineInfo().split(":")[1]));
            }
        }
        int port = 2550;
        if (ports.size() != 0) {
            while (ports.contains(port)) {
                port += 1;
            }
        }
        int id = 1;
        if (ids.size() != 0) {
            while (ids.contains(id)) {
                id += 1;
            }
        }
        ZkUtils.registerEngineInZk(zkClient, id, ip, port, 6000, -1);
        return ConfigFactory.parseString("akka.remote.netty.tcp.port=" + port)
                .withFallback(ConfigFactory.parseString("akka.actor.provider=akka.remote.RemoteActorRefProvider"))
                .withFallback(ConfigFactory.parseString("akka.remote.netty.tcp.hostname=" + ip))
                .withFallback(ConfigFactory.parseString("akka.actor.warn-about-java-serializer-usage=false"))
                .withFallback(ConfigFactory.load());
    }
}
